#pragma once
#include <iostream>
#include <functional>
#include <sys/epoll.h>
#include <assert.h>
#include <unordered_map>
#include <vector>
#include <list>
#include <stdlib.h> 
#include <mutex>
#include <thread>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <memory>
#include <condition_variable>
#include <thread>
#include "log.hpp"

using EventCallBack = std::function<void()>;

class Poller;
class EventLoop;

class Channel
{
public:
    Channel(EventLoop* loop, int fd) :_fd(fd), _loop(loop) {}
    int GetFd() { return _fd; }
    void SetRevents(uint32_t revents) { _revents = revents; }
    uint32_t GetEvents() { return _events; }
    void SetReadCallBack(const EventCallBack& cb)   { _read_callback = cb; }
    void SetWriteCallBack(const EventCallBack& cb)  { _write_callback = cb; }
    void SetCloseCallBack(const EventCallBack& cb)  { _close_callback = cb; }
    void SetErrorCallBack(const EventCallBack& cb)  { _error_callback = cb; }
    void SetEventCallBack(const EventCallBack& cb)  { _event_callback = cb; }
    bool ReadAble()     { return (_events & EPOLLIN); }
    bool WriteAble()    { return (_events & EPOLLOUT); }
    void Update();
    void Remove();

    void EnableRead()   
    {
        _events |= EPOLLIN; 
        Update();
    }

    void EnableWrite()  
    { 
        _events |= EPOLLOUT;
        Update();
    }

    void DisableRead()  
    {
        _events &= ~EPOLLIN; 
        Update();
    }

    void DisableWrite() 
    { 
        _events &= ~EPOLLOUT;
        Update();
    }

    void DisableAll()   
    {
        _events = 0;
        Update();
    }
    
    void HandleEvent()
    {
        if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if(_read_callback) _read_callback();
        }

        if((_revents & EPOLLOUT))
        {
            if(_write_callback) _write_callback();
        }
        else if((_revents & EPOLLHUP))
        {
            if(_close_callback) _close_callback();
        }
        else if((_revents & EPOLLERR))
        {
            if(_error_callback) _error_callback();
        }

        if(_event_callback) _event_callback();
    }

private:
    int _fd;
    EventLoop* _loop;
    uint32_t _events = 0; //当前需要监控的事件
    uint32_t _revents = 0; //当前连接触发的事件
    EventCallBack _read_callback;
    EventCallBack _write_callback;
    EventCallBack _close_callback;
    EventCallBack _error_callback;
    EventCallBack _event_callback;
};


#define MAX_EPOLLEVENTS 1024

// epoll多路转接封装
class Poller
{
private:
    void Update(Channel* channel, int op)
    {
        //int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
        int fd = channel->GetFd();
        struct epoll_event event;
        event.data.fd = fd;
        event.events = channel->GetEvents();
        if(epoll_ctl(_epfd, op, fd, &event)  < 0)
        {
            logMessage(ERROR, "epoll_ctl error, %d:%s", errno, strerror(errno));
        }
    }

    bool HasChannel(Channel* channel)
    {
        if(_channels.find(channel->GetFd()) == _channels.end())
            return false;
        return true;
    }

public:
    Poller()
    {
        _epfd = epoll_create(MAX_EPOLLEVENTS);
        if(_epfd < 0)
        {
            logMessage(FATAL, "epoll_create error, %d:%s", errno, strerror(errno));
            abort();
        }
    }

    //添加或修改监控
    void UpdateEvent(Channel* channel)
    {
        if(HasChannel(channel))
        {
            Update(channel, EPOLL_CTL_MOD); 
             
        }
        else
        {
            _channels.insert(std::make_pair(channel->GetFd(), channel));
            Update(channel, EPOLL_CTL_ADD);
        }
    }
    
    //移除监控
    void RemoveEvent(Channel* channel)
    {
        auto it = _channels.find(channel->GetFd());
        if(it != _channels.end())
            _channels.erase(it);
        Update(channel, EPOLL_CTL_DEL); 
    }

    //开始监控并返回活跃连接
    void Poll(std::vector<Channel*>* active)
    {
        //int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);
        int nfds = epoll_wait(_epfd, _evs, MAX_EPOLLEVENTS, -1);
        if(nfds < 0)
        {
            if(errno == EINTR) // 被信号打断
                return ;
            logMessage(ERROR, "epoll_wait error, %d:%s", errno, strerror(errno));
            abort();
        }

        for(int i = 0; i < nfds; i++)
        {
            auto it = _channels.find(_evs[i].data.fd);
            it->second->SetRevents(_evs[i].events);
            active->push_back(it->second);
        }
    }

private:
    int _epfd;
    struct epoll_event _evs[MAX_EPOLLEVENTS]; 
    std::unordered_map<int, Channel*> _channels;
};


using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;

//定时任务
class TimerTask
{
public:
    TimerTask(uint64_t id, uint64_t timeout, const TaskFunc& task_cb, const ReleaseFunc& release)
        :_id(id), _timeout(timeout), _task_cb(task_cb), _release(release)
    {}

    uint64_t GetTimeout() { return _timeout; }

    void Cancel() { _canceled = true; }

    // 当任务对象被销毁时执行对应的任务
    ~TimerTask()
    {
        if(!_canceled)
            _task_cb();
        _release();
    }

private:
    uint64_t _id; //对象id
    uint64_t _timeout; //超时时间
    bool _canceled = false; //是否被取消
    TaskFunc _task_cb; //任务
    ReleaseFunc _release; //定时任务在到达超时时间后(在时间轮中被移除)，任务对象被销毁，需要在时间轮的查找数组中同步删除该对象(weak_ptr)
};

using WeakTask = std::weak_ptr<TimerTask>;
using PtrTask = std::shared_ptr<TimerTask>;

// 时间轮
class TimerWhell
{
private:
    void RemoveTimerTask(uint64_t id)
    {
        _timers.erase(id); 
    }

    static int CreateTimerFd()
    {
        int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0);
        if(timer_fd < 0)
        {
            logMessage(ERROR, "CreateTimerFd error, %d:%s", errno, strerror(errno)); 
            abort();
        }
        struct itimerspec itime; //设置超时时间为 1s
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;
        timerfd_settime(timer_fd, 0, &itime, nullptr);
        return timer_fd;
    }

    uint64_t ReadTimerFd()
    {
        uint64_t times;
        // 当距离上次的超时时间不足 1s 时，阻塞等待，否则获取超时次数(times)
        //在未阻塞的情况下通常为1次，当执行其他任务的时间过长时，超时次数会超过1次
        if(read(_timerfd, &times, sizeof(times)) < 0) 
        {
            logMessage(ERROR, "ReadTimerFd error, %d:%s", errno, strerror(errno)); 
            abort();
        }
        return times;
    }
    
    // 将_tick向后移动一位，并删除时间轮上该位置的所有任务对象，以执行其析构函数，执行任务
    void Run()
    {
        _tick = (_tick + 1) % _capacity;
        _whell[_tick].clear();
    }

    void OnTime()
    {
        int times = ReadTimerFd(); 
        // ReadTimerFd获取到超时次数，每超时一次为1s，需要执行Run一次
        for(int i = 0 ; i < times ; i++) 
        {
            Run();
        }
    }

    // 新增定时任务
    void TimerAddInLoop(uint64_t id, uint64_t timeout, const TaskFunc& task_cb)
    {
        PtrTask pt(new TimerTask(id, timeout, task_cb, std::bind(&TimerWhell::RemoveTimerTask, this, id))); 
        _timers[id] = WeakTask(pt);
        _whell[(_tick + timeout) % _capacity].push_back(pt); 
    }

    // 定时任务的刷新，即在时间轮中新增一个对应id的任务对象(shared_ptr，只有所有的指针被销毁时才被销毁，执行对应任务)
    bool TimerRefreshInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
            return false;
        PtrTask pt = it->second.lock();
        _whell[(_tick + pt->GetTimeout()) % _capacity].push_back(pt); 
        return true;
    }

    void TimerCancelInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end())
            return ;
        PtrTask pt = it->second.lock();
        if(pt != nullptr)
            pt->Cancel();
    }

public:
    TimerWhell(EventLoop* loop, int capacity = 60)
        :_capacity(capacity), _whell(capacity), _loop(loop), _timerfd(CreateTimerFd()), _timer_channel(new Channel(_loop, _timerfd))
    {
        _timer_channel->SetReadCallBack(std::bind(&TimerWhell::OnTime, this));
        _timer_channel->EnableRead();
    }

    //添加定时任务
    void TimerAdd(uint64_t id, uint64_t timeout, const TaskFunc& task_cb);
    //重置任务超时时间
    void TimerRefresh(uint64_t id);
    //移除定时任务
    void TimerCancel(uint64_t id);

    bool HasTimer(uint64_t id)
    {
        return _timers.find(id) != _timers.end();
    }

private:
    std::vector<std::list<PtrTask>> _whell; //时间轮数组
    int _capacity; //时间轮大小
    int _tick = 0; //时间轮当前位置
    std::unordered_map<uint64_t, WeakTask> _timers; //定时任务集合

    EventLoop* _loop;
    int _timerfd;
    std::unique_ptr<Channel> _timer_channel;
};


using Functor = std::function<void()>;

class EventLoop
{
private:
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC|EFD_NONBLOCK);
        if(efd < 0)
        {
            logMessage(ERROR, "CreateEventFd error, %d:%s", errno, strerror(errno));
            abort();
        } 
        return efd;
    }

    // 将_event_fd清零
    void ReadEventFd()
    {
        uint64_t res = 0;
        if(read(_event_fd, &res, sizeof(res)) < 0)
        {
            if(errno != EINTR)  
            {
                logMessage(ERROR, "ReadEventFd error, %d:%s", errno, strerror(errno));
                abort();
            }
        }
    }

    // 唤醒_event_fd的读事件监控，进而唤醒可能被阻塞的epoll_wait函数，避免任务队列无法得到执行
    void WeakupEventFd()
    {
        uint64_t val = 1;
        if(write(_event_fd, &val, sizeof(val)) < 0)
        {
            if(errno != EINTR && errno != EAGAIN)  
            {
                logMessage(ERROR, "WeakupEventFd error, %d:%s", errno, strerror(errno));
                abort();
            }
        }
    }

public:
    EventLoop() :_thread_id(std::this_thread::get_id())
                ,_event_fd(CreateEventFd())
                ,_event_channel(new Channel(this, _event_fd))
                ,_timerwhell(this)
    {
        // 启动读事件监控，当_event_fd的数据发生改变时(WeakupEventFd函数)，读事件就绪，避免epoll_wait阻塞导致任务队列无法得到执行
        _event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEventFd, this));
        _event_channel->EnableRead();
    }
    
    void RunAllTask()
    {
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(functor);  
        }

        for(auto f: functor)
        {
            f(); 
        }
    }

    void Start()
    {
        while(true)
        {
            //事件监控
            std::vector<Channel*> actives;
            _poller.Poll(&actives);
            //事件处理
            for(auto& channel: actives) 
                channel->HandleEvent();
            //执行函数
            RunAllTask();
        }
    }

    bool IsInLoop() { return _thread_id == std::this_thread::get_id(); }

    void AssertInLoop() { assert(_thread_id == std::this_thread::get_id()); }

    // 将所要执行的函数添加到任务队列中执行，在Start函数中统一执行
    void QueueInLoop(const Functor& cb)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }
        //每次在任务队列中添加函数时，唤醒epoll_wait
        WeakupEventFd();
    }

    // 当前线程与eventloop对应的线程相同则立刻执行函数，否则加入到任务队列中，由eventloop对应的线程执行所有函数，保证线程安全
    void RunInLoop(const Functor& cb)
    {
        if(IsInLoop())
            cb();
        else
            QueueInLoop(cb);
    }

    void UpdateEvent(Channel* channel)  { _poller.UpdateEvent(channel); }            
    void RemoveEvent(Channel* channel)  { _poller.RemoveEvent(channel); }

    void TimerAdd(uint64_t id, uint32_t delay, const TaskFunc& cb)  { _timerwhell.TimerAdd(id, delay, cb); }
    void TimerRefresh(uint64_t id)  { _timerwhell.TimerRefresh(id); }
    void TimerCancel(uint64_t id)   { _timerwhell.TimerCancel(id); }
    bool HasTimer(uint64_t id)  { return _timerwhell.HasTimer(id); }

private:
    std::thread::id _thread_id; // EventLoop与线程一一对应
    int _event_fd; // 计数文件描述符
    Poller _poller; 
    std::unique_ptr<Channel> _event_channel; 
    std::vector<Functor> _tasks; // 任务队列，用于存储QueueInLoop添加的函数
    std::mutex _mutex;
    TimerWhell _timerwhell;
};

void Channel::Update()   { _loop->UpdateEvent(this); }
void Channel::Remove()   { _loop->RemoveEvent(this); }

void TimerWhell::TimerAdd(uint64_t id, uint64_t timeout, const TaskFunc& task_cb)
{
    _loop->RunInLoop(std::bind(&TimerWhell::TimerAddInLoop, this, id, timeout, task_cb));
}

void TimerWhell::TimerRefresh(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWhell::TimerRefreshInLoop, this, id));
}

void TimerWhell::TimerCancel(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWhell::TimerCancelInLoop, this, id));
}

